package com.cupdata.sms.executor;

import com.cupdata.sms.config.ConfigProperties;
import com.cupdata.sms.core.ResizableCapacityLinkedBlockIngQueue;
import com.cupdata.sms.core.ThreadPoolStatus;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Scope;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: zda
 * @description: TODO
 * @date: 2024/4/30 11:04
 */
@Slf4j
@Data
@EnableConfigurationProperties(ConfigProperties.class)
@ConditionalOnProperty(prefix = "channel-core",value = "enableDefaultPool",matchIfMissing = true,havingValue = "true")
@Component("default-thread-pool")
public class CusThreadPoolExecutor implements DisposableBean, ApplicationContextAware, InitializingBean, Ordered {

    private String taskDescribe;
    private Integer corePoolSize;
    private Integer maxPoolSize;
    private Long keepAliveTime = 10L;
    private TimeUnit unit;
    private Integer dequeSize;
    private String poolName;
    private Integer alarmCount;
    private Integer alarmSize;
    private ThreadPoolExecutor threadPoolTaskExecutor;
    private ResizableCapacityLinkedBlockIngQueue<Runnable> blockIngQueue;
    private volatile boolean STOP;
    private final Object poolSizeMonitor;
    private volatile long checkFrequency = 1000;
    private static RejectedExecutionHandler rejectedExecutionHandler =
            new ThreadPoolExecutor.CallerRunsPolicy();

    private static ApplicationContext applicationContext;
    @Autowired
    private ConfigProperties configProperties;

    public CusThreadPoolExecutor() {
        this.unit = TimeUnit.SECONDS;
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        this.corePoolSize = availableProcessors;
        this.maxPoolSize = availableProcessors * 2;
        this.dequeSize = 50000;
        this.STOP = false;
        this.poolSizeMonitor = new Object();
    }

    public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
        CusThreadPoolExecutor.rejectedExecutionHandler = rejectedExecutionHandler;
    }

    @Override
    public void destroy() throws Exception {
        this.shutDown();
        log.info(">>>>>>>>>线程池【{}】销毁成功", this.getPoolName());
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if(configProperties.isEnableDefaultPool()){
            this.setPoolName("default-thread-pool");
            this.start();
        }

    }
    protected void shutDown() {
        synchronized (this.poolSizeMonitor) {
            ThreadPoolExecutor executor = this.getExecutor();
            try {
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                    while (!executor.isTerminated()) {
                        // help gc
                        executor = null;
                    }
                }
            } catch (InterruptedException ie) {
                executor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
    protected void start() {
        this.blockIngQueue = new ResizableCapacityLinkedBlockIngQueue<>(this.dequeSize);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveTime,
                this.unit, this.blockIngQueue,
                new DefaultThreadFactory(this.poolName),
                rejectedExecutionHandler);
        log.info("======= thread-pool {} init success >>>>>>>>>>>>>>>>>>>>>", this.poolName);
        this.threadPoolTaskExecutor = executor;
        log.info(this + "\n");
        // 监控
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
            if (!this.STOP) {
                if (this.alarmSize == null) {
                    this.alarmSize = this.dequeSize;
                }
                if (this.alarmCount == null) {
                    this.alarmCount = this.maxPoolSize;
                }
                int activeCount = this.threadPoolTaskExecutor.getActiveCount();
                int size = this.threadPoolTaskExecutor.getQueue().size();

                if (activeCount <= this.alarmCount && size >= this.alarmSize) {
                    log.warn("【线程池队列告警】线程池状态：{}", this);
                    invokeAlarmPostProcessor();
                }
            }
        }, 0, 1, TimeUnit.SECONDS);

    }

    public ThreadPoolExecutor getExecutor() {
        return this.threadPoolTaskExecutor;
    }

    /**
     * 刷新线程池状态
     */
    public void refresh() {
        synchronized (this.poolSizeMonitor) {
            log.info(">>>>>>>>>>>> 开始刷新线程池参数 >>>>>>>>>>>");
            if (!this.threadPoolTaskExecutor.isShutdown() && !this.threadPoolTaskExecutor.isTerminated()) {
                this.threadPoolTaskExecutor.setCorePoolSize(this.corePoolSize);
                this.threadPoolTaskExecutor.setMaximumPoolSize(this.maxPoolSize);
                this.updateDequeSize(this.dequeSize);
                log.info(">>>>>>>>>>>> 线程池参数刷新完毕 >>>>>>>>>>>");
                log.info(this + "\n");
                this.STOP = false;
            } else {
                throw new RuntimeException("线程池状态异常 不可刷新");
            }
        }
    }

    protected void updateDequeSize(Integer dequeSize) {
        if (this.blockIngQueue == null) {
            throw new RuntimeException(" blockIngQueue is null ");
        } else {
            this.blockIngQueue.setCapacity(dequeSize);
        }
    }

    public void execute(Runnable runnable) {
        ThreadPoolExecutor threadPoolExecutor = this.getExecutor();
        log.info(">>>>>>>>> poolName:{},activeCount：{},queueTaskCount：{}", this.getPoolName(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size());
        threadPoolExecutor.execute(runnable);
    }

    public <E> Future<E> submit(Callable<E> callable) {
        ThreadPoolExecutor executor = this.getExecutor();
        return executor.submit(callable);
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        CusThreadPoolExecutor.applicationContext = applicationContext;
    }

    @Override
    public int getOrder() {
        return 0;
    }


    public static <T> Map<String, T> getBeansOfType(Class<T> tClass) {
        return applicationContext.getBeansOfType(tClass);
    }
    private void invokeAlarmPostProcessor() {
        ThreadPoolStatus threadPoolStatus = this.convertThreadPoolStatus();
        List<AlarmPostProcessor> alarmPostProcessors = obtainAlarmPostProcessorList();
        if (alarmPostProcessors != null) {
            for (AlarmPostProcessor postProcessor : alarmPostProcessors) {
                postProcessor.postProcessor(threadPoolStatus);
            }
        }

    }

    public static List<AlarmPostProcessor> obtainAlarmPostProcessorList() {
        Map<String, AlarmPostProcessor> beansOfType = getBeansOfType(AlarmPostProcessor.class);
        if (CollectionUtils.isEmpty(beansOfType)) {
            return null;
        }
        return new ArrayList<>(beansOfType.values());
    }


    private ThreadPoolStatus convertThreadPoolStatus() {
        ThreadPoolStatus threadPoolStatus = new ThreadPoolStatus();
        threadPoolStatus.setCorePoolSize(this.getCorePoolSize());
        threadPoolStatus.setPoolName(this.getPoolName());
        threadPoolStatus.setActiveCount(this.threadPoolTaskExecutor.getActiveCount());
        threadPoolStatus.setQueueCount(this.threadPoolTaskExecutor.getQueue().size());
        threadPoolStatus.setMaxPoolSize(this.getMaxPoolSize());
        threadPoolStatus.setAlarmSize(this.getAlarmSize());
        threadPoolStatus.setAlarmSize(this.getAlarmCount());
        threadPoolStatus.setTaskDescribe(this.getTaskDescribe());
        threadPoolStatus.setDequeSize(this.getDequeSize());
        return threadPoolStatus;
    }
    static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            this.group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = poolName + "-" + poolNumber.getAndIncrement() + "-thread-";
        }

        public Thread newThread(Runnable r) {
            return new Thread(this.group, r, this.namePrefix + this.threadNumber.getAndIncrement(), 0L);
        }
    }

    public String toString() {
        return "线程池信息{ poolName='" + this.poolName + ", corePoolSize=" + this.corePoolSize + ", maxPoolSize=" + this.maxPoolSize + ", keepAliveTime=" + this.keepAliveTime + ", unit=" + this.unit + ", dequeSize=" + this.dequeSize + ",排队任务数=" + this.threadPoolTaskExecutor.getQueue().size() + ",活跃线程数=" + this.threadPoolTaskExecutor.getActiveCount() + '}';
    }

}
